﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Configuration;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using YiNing.Tools;

namespace YiNing.Tools
{
    public class ReceviceArgs
    {
        string queueName = "";
        EventingBasicConsumer consumer;
        BasicDeliverEventArgs basicDeliverEventArgs;
        RabbitMQClientHandler rabbitMQClient;
        ulong deliveryTag;
        bool autoAck = false;
        int prefetchCount = 30;
        public ReceviceArgs(RabbitMQClientHandler rabbitMQ, string queue, EventingBasicConsumer cu, BasicDeliverEventArgs args, bool autoAck = false, int prefetchCount = 5)
        {
            queueName = queue;
            deliveryTag = args.DeliveryTag;
            rabbitMQClient = rabbitMQ;
            consumer = cu;
            basicDeliverEventArgs = args;
            this.autoAck = autoAck;
            this.prefetchCount = prefetchCount;
        }

        /// <summary>
        /// 手动确认
        /// </summary>
        /// <param name="multiple">是否小于deliveryTag的直接批量确认掉</param>
        /// <returns></returns>
        public bool BasicAck(bool multiple = false)
        {
            try
            {
                if (rabbitMQClient.CheckLiveOrCreate(queueName, out IModel channel, autoAck, prefetchCount)){
                    channel.BasicAck(deliveryTag, multiple);
                }
                return true;
            }
            catch(Exception er)
            {
                return false;
            }
        }

        /// <summary>
        /// 重新丢回队列
        /// </summary>
        /// <returns></returns>
        public bool BasicNackBackQueue()
        {
            try
            {
                if (rabbitMQClient.CheckLiveOrCreate(queueName, out IModel channel, autoAck, prefetchCount)){
                    channel.BasicNack(deliveryTag, false, true);
                }
                return true;
            }
            catch (Exception er)
            {
                if (rabbitMQClient.CheckLiveOrCreate(queueName, out IModel channel, autoAck, prefetchCount)){
                    channel.BasicNack(deliveryTag, false, false);
                }
                return false;
            }
        }

        /// <summary>
        /// 直接丢弃
        /// </summary>
        /// <returns></returns>
        public bool BasicNackAbandon()
        {
            try
            {
                if (rabbitMQClient.CheckLiveOrCreate(queueName, out IModel channel, autoAck, prefetchCount)){
                    channel.BasicNack(deliveryTag, false, false);
                }
                return true;
            }
            catch (Exception er)
            {
                return false;
            }
        }


        public BasicDeliverEventArgs Args
        {
            get { return basicDeliverEventArgs; }
        }
        public EventingBasicConsumer Consumer
        {
            get { return consumer; }
        }
    }
    public class RabbitMQClientHandler
    {
        private bool isPause = false;
        public string defaultRabbitMQHostName = ConfigurationManager.AppSettings["RabbitMQServerHostName"].Trim();
        public string defaultRabbitMQPort = ConfigurationManager.AppSettings["RabbitMQServerPort"].Trim();
        public string defaultRabbitMQUserName = ConfigurationManager.AppSettings["RabbitMQUserName"].Trim();
        public string defaultRabbitMQPassword = ConfigurationManager.AppSettings["RabbitMQPassword"].Trim();
        public string defaultRabbitVirtualHost = ConfigurationManager.AppSettings["RabbitMQVirtualHost"].Trim();
        //public ulong deliveryTag = 0;
        // 定义一个静态变量来保存类的实例
        //private RabbitMQClientHandler uniqueInstance;
        //定义一个标识确保线程同步 
        //private static readonly object locker = new object();

        /*-------------------------------------------------------------------------------------*/

        //ConnectionFactory
        private static ConnectionFactory mc_ConnectionFactory = null;
        //Connection
        private IConnection Connection;

        //发送频道及接收频道分开，避免互相影响，导致整个服务不可用
        //Send Channel
        public IModel SendChannel;
        //Listen Channel
        public ConcurrentDictionary<string, IModel> ListenChannels = new ConcurrentDictionary<string, IModel>();

        //数据监控队列
        public event EventHandler<ReceviceArgs> RabbitmqMessageCallBack;
        public event EventHandler<string> RabbitmqConnectCallback;
        //
        //private SyncDataHandler syncDataHandlerClass;

        /*-------------------------------------------------------------------------------------*/

        /// <summary>
        /// 移除所有注册事件
        /// </summary>
        public void RemoveAllEvent(string eventName = "RabbitmqMessageCallBack")
        {
            var newType = this.GetType();
            foreach (var item in newType.GetEvents())
            {
                FieldInfo _Field = newType.GetField(item.Name, BindingFlags.Instance | BindingFlags.NonPublic);
                if (_Field != null && item.Name.Equals(eventName))
                {
                    object _FieldValue = _Field.GetValue(this);
                    if (_FieldValue != null && _FieldValue is Delegate)
                    {
                        Delegate _ObjectDelegate = (Delegate)_FieldValue;
                        Delegate[] invokeList = _ObjectDelegate.GetInvocationList();
                        if (invokeList != null)
                        {
                            foreach (Delegate del in invokeList)
                            {
                                item.RemoveEventHandler(this, del);
                            }
                        }
                    }
                }
            }
        }

        /// <summary>
        /// 开始连接回调
        /// </summary>
        /// <param name="msg"></param>
        private void StartRabbitmqConnectCallback(string msg)
        {
            if (RabbitmqConnectCallback != null)
            {
                RabbitmqConnectCallback(this, msg);
            }
        }

        /// <summary>
        /// 接受信息回调
        /// </summary>
        /// <param name="msg"></param>
        private void StartRabbitmqMessageCallBack(string queueName, bool autoAck, int prefetchCount, EventingBasicConsumer consumer, BasicDeliverEventArgs re)
        {
            if (RabbitmqMessageCallBack != null)
            {
                RabbitmqMessageCallBack(this, new ReceviceArgs(this, queueName, consumer, re, autoAck, prefetchCount));
            }
        }

        /// <summary>
        /// 定义私有构造函数，使外界不能创建该类实例
        /// </summary>
        public RabbitMQClientHandler()
        {
        }

        public async Task WaitConfirmAsync(TimeSpan timeSpan)
        {
            await Task.Run(() =>
            {
                bool res = SendChannel.WaitForConfirms(timeSpan);
            });
 
        }
        public void StartListing()
        {
            Reconnect();
        }

        ///// <summary>
        ///// 定义公有方法提供一个全局访问点,同时你也可以定义公有属性来提供全局访问点
        ///// </summary>
        ///// <returns></returns>
        //public static RabbitMQClientHandler GetInstance()
        //{
        //    // 当第一个线程运行到这里时，此时会对locker对象 "加锁"，
        //    // 当第二个线程运行该方法时，首先检测到locker对象为"加锁"状态，该线程就会挂起等待第一个线程解锁
        //    // lock语句运行完之后（即线程运行完之后）会对该对象"解锁"
        //    // 双重锁定只需要一句判断就可以了
        //    if (uniqueInstance == null)
        //    {
        //        lock (locker)
        //        {
        //            // 如果类的实例不存在则创建，否则直接返回
        //            if (uniqueInstance == null)
        //            {
        //                uniqueInstance = new RabbitMQClientHandler();
        //            }
        //        }
        //    }
        //    return uniqueInstance;
        //}

        void Connect()
        {
            try
            {
                ListenChannels.Clear();
                LogHelper.WriteLog("获取RabbitMQ服务器参数：" + defaultRabbitMQHostName + ":" + defaultRabbitMQPort + " (" + defaultRabbitMQUserName + "/" + defaultRabbitMQPassword + ")");
                StartRabbitmqConnectCallback("[获取RabbitMQ服务器参数：" + defaultRabbitMQHostName + ":" + defaultRabbitMQPort + " (" + defaultRabbitMQUserName + "/" + defaultRabbitMQPassword + "), 正在连接中...]");
                //连接工厂
                mc_ConnectionFactory = new ConnectionFactory();

                //连接工厂信息
                mc_ConnectionFactory.HostName = defaultRabbitMQHostName;// "localhost";

                int rabbitmq_port = 5672;// 默认是5672端口
                int.TryParse(defaultRabbitMQPort, out rabbitmq_port);
                mc_ConnectionFactory.Port = rabbitmq_port;// "5672"

                mc_ConnectionFactory.UserName = defaultRabbitMQUserName;// "guest";
                mc_ConnectionFactory.Password = defaultRabbitMQPassword;// "guest";
                mc_ConnectionFactory.VirtualHost = defaultRabbitVirtualHost;// "/"

                mc_ConnectionFactory.RequestedHeartbeat = 30;          //心跳包
                mc_ConnectionFactory.AutomaticRecoveryEnabled = true;//自动重连
                mc_ConnectionFactory.TopologyRecoveryEnabled = true;//拓扑重连
                mc_ConnectionFactory.NetworkRecoveryInterval = TimeSpan.FromSeconds(10);

                //创建连接
                Connection = mc_ConnectionFactory.CreateConnection();

                //断开连接时，调用方法自动重连
                Connection.ConnectionShutdown += Connection_ConnectionShutdown;

                SendChannel = Connection.CreateModel();
                //创建发送频道
                //创建接收频道
                //ListenChannels.Add(Connection.CreateModel());

                //发送频道确认模式，发送了消息后，可以收到回应
                SendChannel.ConfirmSelect();

                //if (!string.IsNullOrEmpty(MC_SyncDataConsume))
                //{
                //    //重新监控消息

                //}

                if (RabbitmqConnectCallback != null)
                    RabbitmqConnectCallback(this, "[尝试连接至RabbitMQ服务器：" + defaultRabbitMQHostName + "]");
                LogHelper.WriteLog("尝试连接至RabbitMQ服务器：" + defaultRabbitMQHostName);
            }
            catch (RabbitMQ.Client.Exceptions.BrokerUnreachableException e)
            {
                throw e;
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }

        public void Resume()
        {
            isPause = false;
            Reconnect();
        }

        public void Pause()
        {
            isPause = true;
            Cleanup();
        }
        public void Cleanup()
        {
            try
            {
                if (Connection != null && Connection.IsOpen)
                {
                    try
                    {
                        Connection.Close();
                    }
                    catch (Exception ex)
                    {
                        if (RabbitmqConnectCallback != null)
                            RabbitmqConnectCallback(this, "[RabbitMQ重新连接，正在尝试关闭之前的连接，但遇到错误，等待重新连接...]");
                        LogHelper.WriteLog("RabbitMQ重新连接，正在尝试关闭之前的连接，但遇到错误", ex);
                    }
                    Connection.Close();
                    Connection.Dispose();
                    Connection = null;
                }
                if (SendChannel != null && SendChannel.IsOpen)
                {
                    try
                    {
                        SendChannel.Close();
                    }
                    catch (Exception ex)
                    {
                        LogHelper.WriteLog("RabbitMQ重新连接，正在尝试关闭之前的Channel[发送]，但遇到错误", ex);
                        if (RabbitmqConnectCallback != null)
                            RabbitmqConnectCallback(this, "[RabbitMQ重新连接，正在尝试关闭之前的Channel[发送]，但遇到错误，等待重新连接...]");
                    }
                    SendChannel.Dispose();
                }
                
                foreach (var onekey in ListenChannels.Keys)
                {
                    if (ListenChannels.TryRemove(onekey, out IModel one))
                    {
                        try
                        {
                            one.Close();
                        }
                        catch (Exception ex)
                        {
                            if (RabbitmqConnectCallback != null)
                                RabbitmqConnectCallback(this, "[RabbitMQ重新连接，正在尝试关闭之前的Channel[接收]，但遇到错误，等待重新连接...]");
                            LogHelper.WriteLog("RabbitMQ重新连接，正在尝试关闭之前的Channel[接收]，但遇到错误", ex);
                        }
                        one.Dispose();
                    }
                }
            }
            catch (IOException ex)
            {
                throw ex;
            }
        }

        private void Connection_ConnectionShutdown(object sender, ShutdownEventArgs e)
        {
            LogHelper.WriteLog("):  RabbitMQ已经断开连接，正在尝试重新连接至RabbitMQ服务器");
            if (RabbitmqConnectCallback != null)
                RabbitmqConnectCallback(this, "[):  RabbitMQ已经断开连接，正在尝试重新连接至RabbitMQ服务器...]");
            Reconnect();
        }

        private void Reconnect()
        {
            try
            {

                //清除连接及频道
                Cleanup();

                var mres = new ManualResetEventSlim(false); // state is initially false
                while (!mres.Wait(300)) // loop until state is true, checking every 3s
                {
                    if (isPause)
                    {
                        StartRabbitmqConnectCallback("[暂停中...]");
                        mres.Set();
                    }
                    else
                    {
                        try
                        {
                            StartRabbitmqConnectCallback("[RabbitMQ连接中...]");
                            //连接
                            Connect();

                            mres.Set(); // state set to true - breaks out of loop
                            StartRabbitmqConnectCallback("[连接成功...等待最新的数据]");
                        }
                        catch (Exception ex)
                        {
                            StartRabbitmqConnectCallback("[RabbitMQ尝试连接RabbitMQ服务器出现错误，请检查配置文件是否正确，将在3秒后尝试再次连接...]");
                            LogHelper.WriteLog("RabbitMQ尝试连接RabbitMQ服务器出现错误：" + ex.Message, ex);
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                LogHelper.WriteLog("RabbitMQ尝试重新连接RabbitMQ服务器出现错误：" + ex.Message, ex);
                StartRabbitmqConnectCallback("[RabbitMQ尝试连接RabbitMQ服务器出现错误，请检查配置文件是否正确，将在3秒后尝试再次连接...]");
            }
        }

        /*-------------------------------------------------------------------------------------*/


        /// <summary>
        /// 删除队列
        /// </summary>
        /// <param name="queue"></param>
        /// <param name="ifUnused"></param>
        /// <param name="ifEmpty"></param>
        /// <returns></returns>
        public bool QueueDelete(string queue, bool ifEmpty = true, bool ifUnused = false)
        {
            try
            {
                SendChannel.QueueDelete(queue, ifUnused, ifEmpty);
            }
            catch(Exception er)
            {
                return false;
            }
            return true;
        }

        /// <summary>
        /// 创建路由
        /// </summary>
        /// <param name="exchangeName"></param>
        /// <param name="exchangeType"></param>
        /// <returns></returns>
        public bool ExchangeDeclare(string exchangeName, string exchangeType, bool durable = true, bool autoDelete = false)
        {
            CheckOpened();
            try
            {
                SendChannel.ExchangeDeclarePassive(exchangeName);
                return true;
            }
            catch (Exception er)
            {
                //创建发送频道
                SendChannel = Connection.CreateModel();
                //声明交换机
                SendChannel.ExchangeDeclare(exchangeName, exchangeType, durable, autoDelete);
            }

            return true;
        }

        /// <summary>
        /// 创建队列
        /// </summary>
        /// <param name="queueName"></param>
        /// <param name="durable"></param>
        /// <param name="exclusive"></param>
        /// <param name="autoDelete"></param>
        /// <returns></returns>
        public bool QueueDeclare(string queueName, out QueueDeclareOk queueDeclareOk, bool durable = true, bool exclusive = true, bool autoDelete = false)
        {
            queueDeclareOk = null;
            CheckOpened();
            try
            {
                queueDeclareOk = SendChannel.QueueDeclarePassive(queueName);
                return true;
            }
            catch (Exception er)
            {
                //创建发送频道
                SendChannel = Connection.CreateModel();
                //声明交换机
                queueDeclareOk = SendChannel.QueueDeclare(queueName, durable, exclusive, autoDelete, null);
                
            }
            return true;
        }

        /// <summary>
        /// 获取队列相关
        /// </summary>
        /// <param name="queueName"></param>
        /// <param name="queueDeclareOk"></param>
        /// <returns></returns>
        public bool GetQueueDeclareOk(string queueName, out QueueDeclareOk queueDeclareOk)
        {
            CheckOpened();
            queueDeclareOk = null;
            try
            {
                queueDeclareOk = SendChannel.QueueDeclarePassive(queueName);
                return true;
            }
            catch (Exception er)
            {
                return false;
            }
        }

        /// <summary>
        /// 检测是否开启
        /// </summary>
        /// <returns></returns>
        private bool CheckOpened()
        {
            if (Connection == null || !Connection.IsOpen) { Connection = mc_ConnectionFactory.CreateConnection(); }
            if (SendChannel == null || SendChannel.IsClosed) { SendChannel = Connection.CreateModel(); }
            return true;
        }
        /// <summary>
        /// 绑定路由和队列
        /// </summary>
        /// <returns></returns>
        public bool QueueBind(string exchangeName, string queueName, string routingKey)
        {
            CheckOpened();
            SendChannel.QueueBind(queueName, exchangeName, routingKey, null);
            return true;
        }

        /// <summary>
        /// Fanout路由，发送消息至服务端Fanout Exchange – 不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout交换机转发消息是最快的。 任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的所有Queue上。
        /// </summary>
        /// <param name="queueName">队列名称</param>
        /// <param name="message">消息内容</param>
        /// <returns></returns>
        public bool SendMsg(string exchangeName, string routingKey, string message)
        {
            try
            {
                CheckOpened();
                //设置消息持久性
                IBasicProperties props = SendChannel.CreateBasicProperties();
                props.ContentType = "text/plain";
                props.DeliveryMode = 2;//持久性

                //消息内容转码，并发送至服务器
                var messageBody = System.Text.Encoding.UTF8.GetBytes(message);
                SendChannel.BasicPublish(exchangeName, routingKey, props, messageBody);
                //等待确认
                return true;

            }
            catch (Exception ex)
            {
                LogHelper.WriteLog("RabbitMQ出现通用问题" + ex.Message, ex);

                return false;
            }
        }

        /// <summary>
        /// 检查是否存活
        /// </summary>
        /// <param name="queueName"></param>
        /// <returns></returns>
        public bool CheckLive(string queueName)
        {
            if (ListenChannels.ContainsKey(queueName))
            {
                if (ListenChannels[queueName].IsOpen) return true;
                else { if (ListenChannels.TryRemove(queueName, out IModel value)) { value.Dispose(); return false; } }
            }
            return false;
        }

        /// <summary>
        /// 检查是否存活，如果存活那么直接返回channel,不存活那么就创建一个新的
        /// </summary>
        /// <param name="queueName"></param>
        /// <returns></returns>
        public bool CheckLiveOrCreate(string queueName, out IModel channel, bool autoAck = false, int prefetchCount = 5)
        {
            channel = null;
            if (ListenChannels.ContainsKey(queueName))
            {
                channel = ListenChannels[queueName];
                if (channel.IsOpen) return true;
                else {
                    if (ListenChannels.TryRemove(queueName, out IModel value)) {
                        value.Dispose();
                        RabbitmqMessageConsume(new string[] { queueName }, autoAck, prefetchCount);
                        return CheckLive(queueName);
                    }
                }
            }
            return false;
        }

        public void RabbitmqMessageConsume(string[] queueName, bool autoAck = false, int prefetchCount = 5)
        {
            try
            {
                if (Connection == null || !Connection.IsOpen) { Connection = mc_ConnectionFactory.CreateConnection(); }
             
                foreach (var oneString in queueName)
                {
                    if (ListenChannels.ContainsKey(oneString))
                    {
                        if (ListenChannels[oneString].IsOpen) continue;
                        else { if (ListenChannels.TryRemove(oneString, out IModel value)) { value.Dispose(); } }
                    }
                    var ListenChannel = Connection.CreateModel();
                    //输入1，那如果接收一个消息，但是没有应答，则客户端不会收到下一个消息
                    ListenChannel.BasicQos(0, (ushort)prefetchCount, false);
                    //创建基于该队列的消费者，绑定事件
                    var consumer = new EventingBasicConsumer(ListenChannel);
                    //回应消息监控
                    //consumer.Received += SyncData_Received;
                    consumer.Received += (model, e) =>
                    {
                        try
                        {
                            StartRabbitmqMessageCallBack(oneString, autoAck, prefetchCount, model as EventingBasicConsumer, e);
                        }
                        catch (RabbitMQ.Client.Exceptions.OperationInterruptedException ex1)
                        {
                            Thread.Sleep(50); 
                            if (ListenChannel.IsOpen)
                                ListenChannel.BasicNack(e.DeliveryTag, false, true);
                        }
                        catch (Exception ex)
                        {
                            Thread.Sleep(50);
                            if (ListenChannel.IsOpen)
                                ListenChannel.BasicNack(e.DeliveryTag, false, true);
                        }
                    };
                    //绑定消费者
                    ListenChannel.BasicConsume(oneString, //队列名
                                          autoAck,    //false：手动应答；true：自动应答
                                          consumer);
  

                    if (ListenChannels.TryAdd(oneString, ListenChannel))
                    {
                        StartRabbitmqConnectCallback("[开始监听]" + oneString);
                        LogHelper.WriteLog("开始监控RabbitMQ服务器，队列" + oneString);
                    }
                    else
                    {
                        ListenChannel.Close();
                        ListenChannel.Dispose();
                    }
                }
            }
            catch (Exception ex)
            {
                StartRabbitmqConnectCallback("[监听失败]");
                LogHelper.WriteLog("监听失败", ex);
            }
        }

        /// <summary>
        /// 关闭监听
        /// </summary>
        /// <param name="queueName"></param>
        /// <returns></returns>
        public bool RabbitmqStopListening(string queueName)
        {
            try
            {
                if (ListenChannels.ContainsKey(queueName))
                {
                    ListenChannels[queueName].Close();
                }
                return true;
            }
            catch (Exception ex)
            {
                return false;
            }
        }

        //private static void SyncData_Received(object sender, BasicDeliverEventArgs e)
        //{
        //    try
        //    {
        //        //TOOD 验证程序退出后消费者是否退出去了
        //        var body = e.Body; //消息主体
        //        var message = Encoding.UTF8.GetString(body);

        //        LogHelper.WriteLog("[x] 队列接收到消息：" + message.ToString());

        //        //处理数据
        //        bool processSuccessFlag = new SyncDataHandler().ProcessSyncData(message);
        //        if (processSuccessFlag)
        //        {
        //            //回复确认
        //            ListenChannel.BasicAck(e.DeliveryTag, false);
        //        }
        //        else
        //        {
        //            //未正常处理的消息，重新放回队列
        //            ListenChannel.BasicReject(e.DeliveryTag, true);
        //        }
        //    }
        //    catch (RabbitMQ.Client.Exceptions.OperationInterruptedException ex1)
        //    {
        //        Thread.Sleep(5000);
        //        ListenChannel.BasicNack(e.DeliveryTag, false, true);
        //    }
        //    catch (Exception ex)
        //    {
        //        Thread.Sleep(5000);
        //        ListenChannel.BasicNack(e.DeliveryTag, false, true);
        //    }
        //}
    }
}
